package cn.jly.flink.sideoutput;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * side output demo
 *
 * @author lanyangji
 */
public class SideOutputApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<Device> dataStream = env.socketTextStream("hadoop101", 9999)
                .flatMap(new FlatMapFunction<String, Device>() {
                    @Override
                    public void flatMap(String s, Collector<Device> collector) throws Exception {
                        String[] split = s.split(",");
                        Device device = null;
                        for (int i = 0; i < split.length; i++) {
                            String v = split[i];
                            if (StringUtils.isNoneEmpty(v)) {
                                device = Device.builder().deviceId(Integer.parseInt(split[0].trim()))
                                        .deviceType(split[1].trim())
                                        .timestamp(Long.parseLong(split[2].trim()))
                                        .build();
                            }
                        }

                        collector.collect(device);
                    }
                });

        // 定义 outputTag 的匿名子类
        OutputTag<Device> machineOutputTag = new OutputTag<Device>("machine"){};
        OutputTag<Device> dockerOutputTag = new OutputTag<Device>("docker"){};
        OutputTag<Device> applicationOutputTag = new OutputTag<Device>("application"){};
        OutputTag<Device> middlewareOutputTag = new OutputTag<Device>("middleware"){};

        /*
            side output 要借助
                ProcessFunction
                KeyedProcessFunction
                CoProcessFunction
                ProcessWindowFunction
                ProcessAllWindowFunction
         */
        SingleOutputStreamOperator<Device> processDataStream = dataStream.process(new ProcessFunction<Device, Device>() {
            @Override
            public void processElement(Device value, Context ctx, Collector<Device> out) throws Exception {
                String deviceType = value.getDeviceType();
                switch (deviceType) {
                    case "machine":
                        // 输出到侧输出流
                        ctx.output(machineOutputTag, value);
                        break;
                    case "docker":
                        ctx.output(dockerOutputTag, value);
                        break;
                    case "application":
                        ctx.output(applicationOutputTag, value);
                        break;
                    case "middleware":
                        ctx.output(middlewareOutputTag, value);
                        break;
                    default:
                        // 正常输出
                        out.collect(value);
                }
            }
        });

        // 正常输出
        processDataStream.print("正常输出");
        processDataStream.getSideOutput(machineOutputTag).print("machineOutputTag");
        processDataStream.getSideOutput(dockerOutputTag).print("dockerOutputTag");
        processDataStream.getSideOutput(applicationOutputTag).print("applicationOutputTag");
        processDataStream.getSideOutput(middlewareOutputTag).print("middlewareOutputTag");

        env.execute("SideOutputApp");
    }

    /**
     * 设备实体
     */
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Builder
    private static class Device {
        private Integer deviceId;
        private String deviceType;
        private Long timestamp;
    }
}

